package com.rent.message.utils;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;


@Slf4j
/**
 * 分布式批量处理器 - 纯工具类，不交给Spring管理
 */
public class DistributedBatchProcessor<T> {

    private final List<T> buffer;
    private final int batchSize;
    private final long flushInterval;
    private final Consumer<List<T>> batchConsumer;
    private final String processorName;
    private final Object lock = new Object();

    private ScheduledExecutorService scheduler;
    private volatile boolean running = false;

    /**
     * 构造方法
     */
    public DistributedBatchProcessor(String processorName, int batchSize, long flushInterval,
                                     Consumer<List<T>> batchConsumer) {
        this.processorName = processorName;
        this.batchSize = batchSize;
        this.flushInterval = flushInterval;
        this.batchConsumer = batchConsumer;
        this.buffer = Collections.synchronizedList(new ArrayList<>());
    }

    /**
     * 启动处理器
     */
    public void start() {
        if (running) {
            return;
        }

        this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread thread = new Thread(r, "BatchProcessor-" + processorName);
            thread.setDaemon(true);
            return thread;
        });

        this.scheduler.scheduleAtFixedRate(this::scheduledFlush,
                flushInterval, flushInterval, TimeUnit.MILLISECONDS);

        this.running = true;
        log.info("分布式批量处理器 [{}] 已启动, 批量大小: {}, 刷新间隔: {}ms",
                processorName, batchSize, flushInterval);
    }

    /**
     * 添加数据到批量处理器
     */
    public void add(T item) {
        if (!running) {
            throw new IllegalStateException("批量处理器未启动");
        }

        synchronized (lock) {
            buffer.add(item);
            if (buffer.size() >= batchSize) {
                flushImmediately();
            }
        }
    }

    /**
     * 立即刷新
     */
    public void flushImmediately() {
        synchronized (lock) {
            if (!buffer.isEmpty()) {
                List<T> toProcess = new ArrayList<>(buffer);
                buffer.clear();
                processBatch(toProcess);
            }
        }
    }

    /**
     * 定时刷新
     */
    private void scheduledFlush() {
        synchronized (lock) {
            log.debug("处理器 [{}] 定时刷新", processorName);
            if (!buffer.isEmpty()) {
                List<T> toProcess = new ArrayList<>(buffer);
                buffer.clear();
                processBatch(toProcess);
            }
        }
    }

    /**
     * 处理批量数据
     */
    private void processBatch(List<T> batch) {
        if (batch.isEmpty()) {
            log.debug("处理器 [{}] 批量处理空数据", processorName);
            return;
        }

        try {
            batchConsumer.accept(batch);
            log.debug("处理器 [{}] 批量处理了 {} 条数据", processorName, batch.size());
        } catch (Exception e) {
            log.error("处理器 [{}] 批量处理失败，将数据重新放回缓冲区", processorName, e);
            synchronized (lock) {
                buffer.addAll(0, batch);
            }
        }
    }

    /**
     * 获取当前缓冲区大小
     */
    public int getBufferSize() {
        return buffer.size();
    }

    /**
     * 停止处理器
     */
    public void shutdown() {
        if (!running) {
            return;
        }

        running = false;
        flushImmediately();

        if (scheduler != null) {
            scheduler.shutdown();
            try {
                if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                    scheduler.shutdownNow();
                }
            } catch (InterruptedException e) {
                scheduler.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }

        log.info("分布式批量处理器 [{}] 已停止", processorName);
    }
}